//
// Created by dave on 18-8-7.
//

#pragma once

#include <atomic>
#include <memory>
#include <vector>
#include <random>
#include <thread>
#include <glog/logging.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/net/EventLoopThreadPool.h>
#include <muduo/net/TcpClient.h>
#include "codec/data_pack.h"
#include "codec/data_utils.h"
#include "codec/terminal/term_register.h"
#include "codec/terminal/location_report.h"

using namespace muduo;
using namespace muduo::net;

class Jtt808Emulator {
public:
    Jtt808Emulator(EventLoop* loop,
                    const InetAddress& serverAddr,
                    int threadCount,
                    int connectPerThread)
    : threadPool_(loop, "emulator"), threadCount_(threadCount),
    connectPerThread_(connectPerThread), serverAddr_(serverAddr)
    {
        threadPool_.setThreadNum(threadCount);
    }

    void connect() {
        threadPool_.start();

        for (int i = 0; i < threadCount_ * connectPerThread_; i++) {
            // std::string name = "emulator" + std::to_string(i + 1);
            auto c = std::make_shared<TcpClient>(threadPool_.getNextLoop(), serverAddr_, "emulator");
            c->setConnectionCallback(std::bind(&Jtt808Emulator::onConnection, this, std::placeholders::_1));
            c->setMessageCallback(std::bind(&Jtt808Emulator::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            c->connect();
            clients_.push_back(c);
            std::this_thread::sleep_for(std::chrono::milliseconds(2));
        }
    }

    void start() {
        std::random_device rand;
        std::default_random_engine engine(rand());
        std::uniform_int_distribution<> dis(0, 10000);
        auto dice = std::bind(dis, engine);

        for (int i = 0; i < threadCount_; i++) {
            std::thread th([=, &dice, gid = i] {
                if (getConnected() < threadCount_ * connectPerThread_) {
                    std::this_thread::sleep_for(std::chrono::seconds(1));
                }

                LOG(INFO) << "3秒钟后开始发送数据";
                std::this_thread::sleep_for(std::chrono::seconds(3));
                auto startTime = std::chrono::system_clock::now();
                unsigned short seq = 1;

#pragma clang diagnostic push
#pragma clang diagnostic ignored "-Wmissing-noreturn"
                while (true) {
                    for (int j = gid * connectPerThread_; j < connectPerThread_ * (gid + 1); j++) {
                        TcpConnectionPtr conn = clients_[j]->connection();
                        std::string sim = string_format("0188%08d", j + 1);
                        DataPack re;
                        re.setMsgId(0x0100); // 注册
                        re.setSeqNo(seq);
                        re.setSim(sim);

                        auto body = std::make_shared<TermRegister>();
                        body->setProvince(42);
                        body->setCity(100);
                        body->setMaker("abcde");
                        body->setModel("12345678");
                        body->setDeviceId("abcdefg");
                        body->setColor(2);
                        body->setLicense("鄂A12345");
                        re.setBody(body);
                        std::vector<char> bytes = re.encode();
                        conn->send(bytes.data(), static_cast<int>(bytes.size()));

                        re.setMsgId(0x0200); // 上报位置
                        re.setSeqNo(static_cast<unsigned short>(seq + 1));
                        auto loc = std::make_shared<LocationReport>();
                        loc->setAlarm(0);
                        loc->setFlag(2);
                        loc->setLat(30 + dice() / 10000.0);
                        loc->setLng(118 + dice() / 10000.0);
                        loc->setHeight(static_cast<unsigned short>(0.15 * dice()));
                        loc->setSpeed(static_cast<unsigned short>(0.08 * dice()));
                        loc->setDirection(static_cast<unsigned short>(0.036 * dice()));
                        loc->setTime(std::time(nullptr));
                        re.setBody(loc);
                        bytes = re.encode();
                        conn->send(bytes.data(), static_cast<int>(bytes.size()));

                        long c = sendCounter_ += 2;
                        if (c % 20000 == 0) {
                            auto end = std::chrono::system_clock::now();
                            auto diff = std::chrono::duration_cast<std::chrono::milliseconds>(end - startTime);
                            LOG(INFO) << "已发出（万包）：" << c / 10000 << ", " << diff.count();
                        }
                    }

                    seq += 2;
                    if (++seq >= 0xFFFF - 1) {
                        seq = 1;
                    }

                    // std::this_thread::sleep_for(std::chrono::milliseconds(10));
                }
#pragma clang diagnostic pop
            });

            th.detach();
        }

    }

    std::vector<std::shared_ptr<TcpClient>>& getClients() {
        return clients_;
    }

    int getConnected() {
        return connected_.load();
    }

private:
    void onConnection(const TcpConnectionPtr& conn) {
        if (conn->connected()) {
            int c = ++connected_;
            conn->setTcpNoDelay(true);
            LOG(INFO) << "已连接：" << c;
        } else {
            int c = --connected_;
            LOG(INFO) << "断开连接：" << c;
        }
    }

    void onMessage(const TcpConnectionPtr& conn, Buffer* buf, Timestamp ts) {
        size_t size = buf->readableBytes();
        if (size > 4 * 1024 * 1024) {
            buf->retrieveAll();
            return;
        }

        const char* p = buf->peek();
        size_t start = _find_mark(p, 0, size);
        if (start == 0 && p[0] != 0x7E) {
            buf->retrieveAll();
        } else {
            do {
                size_t end = _find_mark(p, start + 1, size);
                if (end > 0) {
                    if (end - start == 1) {
                        start = end;
                        continue;
                    }

                    onFrame(conn, p + start, end - start + 1);
                    start = _find_mark(p, end + 1, size);
                    if (start == 0) {
                        buf->retrieve(end + 1);
                        break;
                    }
                } else {
                    buf->retrieve(start);
                    break;
                }
            } while (true);
        }
    }

    void onFrame(const TcpConnectionPtr& conn, const char *ptr, size_t length) {
        int64_t count = ++receiveCounter_;
        if (count % 20000 == 0) {
            LOG(INFO) << "接收(万包)：" << count / 10000;
        }

        if (length < 14) {
            LOG(WARNING) << "无效桢：" << length << "字节";
            return;
        }

        DataPack dp = decode_data_pack(ptr, 0, length);
        // LOG(INFO) << "接收：" << std::hex << dp.getMsgId() << ":" << std::dec << dp.getSeqNo() << ":" << dp.getSim();
    }

    size_t _find_mark(const char* ptr, size_t start, size_t stop) {
        for (size_t i = start; i < stop; i++) {
            if (ptr[i] == 0x7E) {
                return i;
            }
        }
        return 0;
    }

private:
    int threadCount_;
    int connectPerThread_;
    InetAddress serverAddr_;
    std::atomic_int connected_{0};
    std::atomic_int64_t sendCounter_{0};
    std::atomic_int64_t receiveCounter_{0};
    EventLoopThreadPool threadPool_;
    std::vector<std::shared_ptr<TcpClient>> clients_;
};

